package routeinfo

import (
	"bbq/common"
	"bbq/common/namesrv"
	"bbq/common/protocol/route"
	"bbq/namesrv/namesrvLog"
	"github.com/panjf2000/gnet"
	"sync"
	"time"
)

type BrokerLiveInfo struct {
	LastUpdateTimestamp int64 //最后一次跟新的时间戳

	DataVersion int64 //数据版本号

	Conn gnet.Conn //链接

	//master broker打开的用于主从通信的IP:PORT信息，在接收broker发来的注册请求时
	//如果发现是slave broker，则会找到其对应主master的haServerAddr信息，并返回给slave broker
	//slave broker根据该地址与master建立连接，之后的消息同步以及其他数据的同步就在此连接上进行
	HaServerAddr string
}

type RouteInfoManager struct {
	TopicQueueTable  map[string] /*topic*/ []*route.QueueData     //通过这个可以获取主题所在的broker以及队列个数信息
	BrokerAddrTable  map[string] /*brokerName*/ *route.BrokerData //通过这个可以获取broker所在集群、broker的ip端口地址
	ClusterAddrTable map[string] /*clusterName*/ map[string]int8  /*brokerName*/ //集群名对应的broker名集合 可以通过集群名查询到该集群下有哪些brokerName
	BrokerLiveTable  map[string] /*brokerAddr*/ *BrokerLiveInfo   //broker心跳信息 broker地址对应broker心跳信息

	sync.RWMutex
}

/*
 * 注册broker信息
 */
func (rm *RouteInfoManager) RegisterBroker(clusterName string, brokerAddr string, brokerName string, brokerId int8,
	haServerAddr string, topicConfigTable map[string] /*topicName*/ common.TopicConfig, dataVersion int64, conn gnet.Conn) *namesrv.RegisterBrokerResult {
	defer rm.Unlock()
	rm.Lock()

	result := &namesrv.RegisterBrokerResult{}

	brokerNames := rm.ClusterAddrTable[clusterName]
	if brokerNames == nil {
		brokerNames = make(map[string]int8, 8)
	}
	brokerNames[brokerName] = 0
	rm.ClusterAddrTable[clusterName] = brokerNames

	brokerData := rm.BrokerAddrTable[brokerName]

	if brokerData == nil {
		brokerData = &route.BrokerData{
			BrokerName:  brokerName,
			Cluster:     clusterName,
			BrokerAddrs: make(map[int8] /*brokerId*/ string, 8),
		}
		rm.BrokerAddrTable[brokerName] = brokerData
	}

	brokerData.BrokerAddrs[brokerId] = brokerAddr

	//主broker跟新队列信息
	if brokerId == 0 && topicConfigTable != nil {
		if rm.isBrokerTopicConfigChanged(brokerAddr, dataVersion) {
			for _, topicConfig := range topicConfigTable {
				rm.createAndUpdateQueueData(brokerName, topicConfig)
			}
		}
	}

	//跟新 BrokerLiveInfo 心跳信息
	rm.BrokerLiveTable[brokerAddr] = &BrokerLiveInfo{
		LastUpdateTimestamp: time.Now().UnixNano() / 1e6,
		DataVersion:         dataVersion,
		Conn:                conn,
		HaServerAddr:        haServerAddr,
	}

	if brokerId != 0 {
		masterAddr := brokerData.BrokerAddrs[0]
		if len(masterAddr) != 0 {
			brokerLiveInfo := rm.BrokerLiveTable[masterAddr]
			if brokerLiveInfo != nil {
				result.HaServerAddr = brokerLiveInfo.HaServerAddr
				result.MasterAddr = masterAddr
			}
		}
	}

	return result
}

func (rm *RouteInfoManager) isBrokerTopicConfigChanged(brokerAddr string, dataVersion int64) bool {
	prevLiveInfo := rm.BrokerLiveTable[brokerAddr]
	if prevLiveInfo == nil {
		return true
	}
	return prevLiveInfo.DataVersion != dataVersion
}

/**
 * 跟新队列信息
 */
func (rm *RouteInfoManager) createAndUpdateQueueData(brokerName string, topicConfig common.TopicConfig) {
	queueData := &route.QueueData{
		BrokerName: brokerName,
		QueueNum:   topicConfig.QueueNum,
		TopicName:  topicConfig.TopicName,
	}

	queueDataList := rm.TopicQueueTable[topicConfig.TopicName]

	if queueDataList == nil {
		queueDataList = make([]*route.QueueData, 0)
		queueDataList = append(queueDataList, queueData)
		rm.TopicQueueTable[topicConfig.TopicName] = queueDataList
		namesrvLog.Log.Infof("new topic registered %s %v", topicConfig.TopicName, queueData)
		return
	}

	addNewOne := true
	removeIdx := -1
	for i, data := range queueDataList {
		if data.BrokerName == brokerName {
			if *data == *queueData {
				addNewOne = false
			} else {
				removeIdx = i
			}
			break
		}
	}

	if removeIdx != -1 {
		queueDataList = append(queueDataList[:removeIdx], queueDataList[removeIdx+1:]...)
	}

	if addNewOne {
		queueDataList = append(queueDataList, queueData)
	}
}
